package com.flume.JSONFilter;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

public class JsonFieldFilter implements Interceptor {
	
	private static final Logger logger = LoggerFactory.getLogger(JsonFieldFilter.class);
	
	private String filterKeys;
	private String encoding;
	private List<String> keys;
	
	public JsonFieldFilter(String filterKeys, String encoding) {
		this.filterKeys = filterKeys;
		this.encoding = encoding;
	}
	
	public void initialize() {
		// TODO Auto-generated method stub
		keys = Arrays.asList(filterKeys.split(","));
		logger.info("filter keys : {}", keys);
	}

	public Event intercept(Event event) {
		// TODO Auto-generated method stub

		try {
			JSONObject body = new JSONObject();
			JSONObject eventBody = JSONObject.parseObject(new String(event.getBody(), encoding));
			for(String key:keys) {
				if(eventBody.containsKey(key)) {
					body.put(key, eventBody.get(key));
				}else {
					body.put(key, "NULL");
				}
			}
			event.setBody(body.toString().getBytes(encoding));
			
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			logger.error("unsupported encoding : {}",encoding);
			e.printStackTrace();
		}		
		
		return event;
	}

	public List<Event> intercept(List<Event> events) {
		// TODO Auto-generated method stub
		List<Event> result = new ArrayList<Event>();
		for(Event e:events) {
			result.add(intercept(e));
		}
		return result;
	}

	public void close() {
		// TODO Auto-generated method stub
		
	}

	/**
	 * 
	 * @author lenovo
	 * @describe key 需要过滤的key值
	 * 			 encoding 文件编码
	 * 
	 */
	public static class Builder implements Interceptor.Builder{
		// configure parameters
		public static String FilterKeys = "key";
		public static String Encoding = "encoding";
				
		private String encoding;
		private String filterKeys;
		
		public void configure(Context context) {
			// TODO Auto-generated method stub
			
			// key need to filter
			filterKeys = context.getString(FilterKeys, null);
			// message encoding,default utf8
			encoding = context.getString(Encoding,"UTF-8");
		}

		public Interceptor build() {
			// TODO Auto-generated method stub
			return new JsonFieldFilter(filterKeys, encoding);
		}
		
	}
	
}
